// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/connectors/hive/storage_adapters/s3fs/RegisterS3FileSystem.h>
#include <pollux/connectors/hive/storage_adapters/s3fs/tests/S3Test.h>

namespace kumo::pollux::filesystems {
namespace {

std::string cacheKeyFunc(
    std::shared_ptr<const config::ConfigBase> config,
    std::string_view path) {
  return config->get<std::string>("hive.s3.endpoint").value();
}

class S3FileSystemRegistrationTest : public S3Test {
 protected:
  static void SetUpTestCase() {
    memory::MemoryManager::testingSetInstance({});
    filesystems::registerS3FileSystem(cacheKeyFunc);
  }

  static void TearDownTestCase() {
    filesystems::finalizeS3FileSystem();
  }
};
} // namespace

TEST_F(S3FileSystemRegistrationTest, readViaRegistry) {
  const char* bucketName = "data2";
  const char* file = "test.txt";
  const std::string filename = localPath(bucketName) + "/" + file;
  const std::string s3File = s3URI(bucketName, file);
  addBucket(bucketName);
  {
    LocalWriteFile writeFile(filename);
    writeData(&writeFile);
  }
  auto hiveConfig = minioServer_->hiveConfig();
  {
    auto s3fs = filesystems::getFileSystem(s3File, hiveConfig);
    auto readFile = s3fs->openFileForRead(s3File);
    readData(readFile.get());
  }
}

TEST_F(S3FileSystemRegistrationTest, fileHandle) {
  const char* bucketName = "data3";
  const char* file = "test.txt";
  const std::string filename = localPath(bucketName) + "/" + file;
  const std::string s3File = s3URI(bucketName, file);
  addBucket(bucketName);
  {
    LocalWriteFile writeFile(filename);
    writeData(&writeFile);
  }
  auto hiveConfig = minioServer_->hiveConfig();
  FileHandleFactory factory(
      std::make_unique<SimpleLRUCache<std::string, FileHandle>>(1000),
      std::make_unique<FileHandleGenerator>(hiveConfig));
  auto fileHandleCachePtr = factory.generate(s3File);
  readData(fileHandleCachePtr->file.get());
}

TEST_F(S3FileSystemRegistrationTest, cacheKey) {
  auto hiveConfig = minioServer_->hiveConfig();
  auto s3fs = filesystems::getFileSystem(kDummyPath, hiveConfig);
  std::string_view kDummyPath2 = "s3://dummy2/foo.txt";
  auto s3fs_new = filesystems::getFileSystem(kDummyPath2, hiveConfig);
  // The cacheKeyFunc function allows fs caching based on the endpoint value.
  ASSERT_EQ(s3fs, s3fs_new);
}

TEST_F(S3FileSystemRegistrationTest, finalize) {
  auto hiveConfig = minioServer_->hiveConfig();
  auto s3fs = filesystems::getFileSystem(kDummyPath, hiveConfig);
  POLLUX_ASSERT_THROW(
      filesystems::finalizeS3FileSystem(),
      "Cannot finalize S3FileSystem while in use");
}
} // namespace kumo::pollux::filesystems
